package com.atguigu.edu.realtime.dim.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.TableProcessDim;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.FlinkSourceUtil;
import com.atguigu.edu.realtime.common.util.HBaseUtil;
import com.atguigu.edu.realtime.dim.function.DimSinkFunction;
import com.atguigu.edu.realtime.dim.function.TableProcessFunction;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

/**
 * Dim层处理
 *
 */

public class DimApp extends BaseApp {
    public static void main(String[] args) {
        new DimApp().start(10001,4,"dim_app", Constant.TOPIC_DB);
    }


    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //TODO 对流中的数据进行处理，并进行简单的ETL jsonStr -》jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDs = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {

                        //需要对异常转换进行捕捉
                        try {
                            JSONObject jsonObject = JSON.parseObject(jsonStr);
                            String database = jsonObject.getString("database");
                            String type = jsonObject.getString("type");
                            String data = jsonObject.getString("data");
                            //数据清洗
                            if ("edu_realtime".equals(database) && ("insert".equals(type) || "delete".equals(type) || "bootstrap-insert".equals(type)) && data.length() > 2 && data != null) {
                                //收集数据
                                collector.collect(jsonObject);
                            }
                        } catch (Exception e) {
                            throw new RuntimeException("不是一个标准Json");
                        }
                    }
                }
        );
        //jsonObjDs.print();
        //2> {"database":"edu_realtime","xid":2470454,"data":{"course_id":435,"deleted":"0","create_time":"2024-07-06 11:35:30","user_id":551,"position_sec":689,"id":15498,"chapter_id":21400},"commit":true,"type":"insert","table":"user_chapter_process","ts":1720236930}
        //1> {"database":"edu_realtime","xid":2470463,"data":{"course_id":435,"deleted":"0","create_time":"2024-07-06 11:35:31","user_id":551,"comment_txt":"评论 123123","id":15439,"chapter_id":21400},"commit":true,"type":"insert","table":"comment_info","ts":1720236930}

        //TODO 使用FlinkCDC从配置表中中读取配置信息
        //创建mysqlSource数据源
        MySqlSource<String> mysqlSource = FlinkSourceUtil.getMySqlSource("edu_realtime_config", "dim_table_process");

        //读取数据源数据封装为流
        //必须设置并行度为1，否则不能封装为流
        DataStreamSource<String> mysqlStrDs = env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "mysql_source").setParallelism(1);

        //mysqlStrDs.print("FlinkCDC");
        //FlinkCDC:1> {"before":null,"after":{"source_table":"test_point_question","sink_table":"dim_test_point_question","sink_family":"info","sink_columns":"id,point_id,question_id,create_time,publisher_id,deleted","sink_row_key":"id"},"source":{"version":"1.9.7.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,"snapshot":"false","db":"edu_realtime_config","sequence":null,"table":"dim_table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1720176935373,"transaction":null}

        //TODO 对配置流中的数据 进行类型转换 jsonStr->实体类对象
        SingleOutputStreamOperator<TableProcessDim> tpDs  = mysqlStrDs.map(
                new MapFunction<String, TableProcessDim>() {
                    @Override
                    public TableProcessDim map(String jsonStr) throws Exception {
                        //字符串转换为json对象
                        JSONObject jsonObject = JSON.parseObject(jsonStr);
                        //获取对配置表进行的操作的类型
                        String op = jsonObject.getString("op");
                        TableProcessDim tableProcessDim = null;
                        if ("d".equals(op)) {
                            //从配置表删除了一条数据
                            tableProcessDim = jsonObject.getObject("before", TableProcessDim.class);
                        } else {
                            //说明配置表做了读取，更新，添加操作 这些需要从after属性中获取对应的配置信息
                            tableProcessDim = jsonObject.getObject("after", TableProcessDim.class);
                        }
                        //把op属性加入tableProcessDim
                        tableProcessDim.setOp(op);
                        return tableProcessDim;
                    }
                }
        ).setParallelism(1);
        tpDs.print();


        //TODO 根据读取到的配置信息，在Hbase中建表或者删除，流中的信息不做更改
        SingleOutputStreamOperator<TableProcessDim> hbaseDs = tpDs.map(
                new RichMapFunction<TableProcessDim, TableProcessDim>() {
                    private Connection hbaseConnection;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hbaseConnection = HBaseUtil.getHbaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hbaseConnection);
                    }

                    @Override
                    public TableProcessDim map(TableProcessDim tableProcessDim) throws Exception {
                        //获取对配置表进行的操作类型
                        String op = tableProcessDim.getOp();
                        //获取表名
                        String sinkTable = tableProcessDim.getSinkTable();
                        String[] sinkFamilies = tableProcessDim.getSinkFamily().split(",");
                        if ("d".equals(op)) {
                            //从配置表中删除了一条数据 将这条配置信息对应的维度表从Hbase中删除掉
                            HBaseUtil.dropHBaseTable(hbaseConnection, Constant.HBASE_NAMESPACE, sinkTable);
                        } else if ("r".equals(op) || "c".equals(op)) {
                            //从配置表中读取了一条配置或者向配置表中添加了一条配置 在Hbase中创建对象的维度表
                            HBaseUtil.createHBaseTable(hbaseConnection, Constant.HBASE_NAMESPACE, sinkTable, sinkFamilies);

                        } else {
                            //对配置表进行的更新操作 先从Hbase中删除对应的维度表，再创建
                            //将这条配置信息对应的维度表从Hbase中删除掉
                            HBaseUtil.dropHBaseTable(hbaseConnection, Constant.HBASE_NAMESPACE, sinkTable);
                            //创建维度表
                            HBaseUtil.createHBaseTable(hbaseConnection, Constant.HBASE_NAMESPACE, sinkTable, sinkFamilies);
                        }


                        return tableProcessDim;
                    }
                }
        );
        hbaseDs.print();

       //TODO 对配置流进行广播
        MapStateDescriptor<String, TableProcessDim> mapStateDescriptor = new MapStateDescriptor<>("mapStateDescriptor", String.class, TableProcessDim.class);
        BroadcastStream<TableProcessDim> broadcastDs = hbaseDs.broadcast(mapStateDescriptor);


        //TODO 将主流和广播流关联
        BroadcastConnectedStream<JSONObject, TableProcessDim> connectDs
                = jsonObjDs.connect(broadcastDs);


        //TODO 对关联后的数据进行处理，过滤出维度数据
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> dimDs = connectDs.process(
                new TableProcessFunction(mapStateDescriptor)
        );
        dimDs.print();

        //TODO 将维度数据同步到Hbase对应的表中
        dimDs.addSink(new DimSinkFunction());
        dimDs.print("dim");



    }
}
